package com.uni.hbase.utils;

import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceExistException;
import org.apache.hadoop.hbase.TableExistsException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.namespace.NamespaceAuditor;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/*
 * @Author: Uni
 * @Time: 2022/5/3
 * @TODO 实现 Hbase的 DDL 操作
 */
public class HbaseDDL {

    /**
     * 获取所有的表名 类似 scan
     * @return 获取所有表名
     */
    public static List<String> scanTables(){
        List<String> tableNames = new LinkedList<>();
        try {
            for (TableName tableName : HbaseConn.getAdmin().listTableNames()) {
                tableNames.add(tableName.getNameAsString());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return tableNames;
    }
    /**
     * 根据表名判断表是否存在
     * @param tableName 表名
     * @return 判断的结果, True说明表已经存在, 反之则不存在
     */
    public static boolean tableExists(String tableName){
        boolean existed = false;
        try{
            existed = HbaseConn.getAdmin().tableExists(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return existed;
    }


    /**
     * 根据表名和列族参数列表创建表
     * @param tableName 表名
     * @param columnsFamilies 列族的可变参数
     * @return  是否创建成功
     * @throws TableExistsException 表已经存在的异常
     */
    public static boolean createTable(String tableName, String ... columnsFamilies)
            throws TableExistsException {
        return createTableAndNeed(tableName, columnsFamilies) != null;
    }
    /**
     * 创建表
     * @param tableName 表名
     * @param cls   列族信息
     * @return 创建的表对象, 若创建失败则为 Null
     */
    public static Table createTableAndNeed(String tableName, String... columnFamilies)
            throws TableExistsException {
        // 过滤条件 1: 列族不能为空
        if (columnFamilies.length == 0)
            throw new NullPointerException("缺少列族参数");
        // 过滤条件2 : 表规定为不存在
        if(tableExists(tableName))
            throw new TableExistsException("[ " + tableName+" ]表已存在!");
        TableName tableNamePro = TableName.valueOf(tableName);
        // 1. 构建列族的描述器
        List<ColumnFamilyDescriptor> list = new LinkedList<>();
        // 2. 封装列族描述器, 遍历列族，（描述器由构造器构建）
        for (String cf : columnFamilies) {
            list.add(ColumnFamilyDescriptorBuilder.
                    newBuilder(Bytes.toBytes(cf)).build());
        }
        // 3. 封装表描述器
        TableDescriptor tableDescriptor =
                TableDescriptorBuilder.
                        newBuilder(tableNamePro).
                            setColumnFamilies(list).build();
        // 4. 创建表
        try {
            HbaseConn.getAdmin().createTable(tableDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 5. 获取表
        Table table = null;
        try {
            table = HbaseConn.getConn().getTable(tableNamePro);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return table;
    }

    /**
     * 判断命名空间是否存在
     * @param namespace 命名空间
     * @return 存在则返回 true
     */
    public static boolean existedNamespace(String namespace){
        try {
            NamespaceDescriptor[] descriptors = HbaseConn.getAdmin().listNamespaceDescriptors();
            for (NamespaceDescriptor descriptor : descriptors) {
                if(descriptor.getName().equals(namespace))
                    return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 创建命名空间
     * @param namespace 命名空间
     */
    public static void createNamespace(String namespace){
        // 创建命名描述器
        NamespaceDescriptor descriptor = NamespaceDescriptor.create(namespace).build();
        try {
            HbaseConn.getAdmin().createNamespace(descriptor);
        } catch (IOException e) {
            if(e instanceof NamespaceExistException)
                System.out.println("[FAILED] >> 命名【 " + namespace +" 】已存在");
            e.printStackTrace();
        }
    }

    public static Map<String, String> descNamespace(String namespace){
        Map<String, String> prop = null;
        try {
            NamespaceDescriptor descriptor = HbaseConn.getAdmin().getNamespaceDescriptor(namespace);
            prop = descriptor.getConfiguration();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return prop;
    }
    /**
     * 获取所有命名空间的哈希信息: K=命名空间名称 V=该命名空间的配置
     * @return 命名空间的信息
     */
    public static Map<String, Map<String, String>>  listNamespaceName(){
        Map<String, Map<String, String>> namespaces = new HashMap<>();
        // 根据获取的描述器来查询命名空间
        try {
            for (NamespaceDescriptor namespaceDescriptor :
                                        HbaseConn.getAdmin().
                                             listNamespaceDescriptors()) {
                namespaces.put(namespaceDescriptor.getName(),
                                    namespaceDescriptor.getConfiguration());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return namespaces;
    }

    /**
     * 修改指定命名空间的信息
     * @param namespace 命名空间名称
     * @param properties    命名空间的配置
     * @return  配置的结果，成功则返回 true
     */
    public static boolean modifyOneNamespace(String namespace, Map<String,String> properties){
        Admin admin = HbaseConn.getAdmin();
        NamespaceDescriptor namespaceDescriptor = null;
        try {
            namespaceDescriptor = admin.getNamespaceDescriptor(namespace);
            if(namespaceDescriptor != null) {
                // 插入属性
                for (Map.Entry<String, String> entry : properties.entrySet()) {
                    namespaceDescriptor.setConfiguration(entry.getKey(), entry.getValue());
                }
                // 通过 Admin 更新命名空间
                admin.modifyNamespace(namespaceDescriptor);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除指定的命名空间
     * @param namespace
     * @return 删除成功则返回 true
     */
    public static boolean deleteNamespace(String namespace){
        try {
            HbaseConn.getAdmin().deleteNamespace(namespace);
            return !existedNamespace(namespace);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
}
